package com.we.risk.phoneRecord.stage2.dimuser;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.userlabelplatform.table.TableUserAccount;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import com.weicai.caesar.CaesarUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Deprecated
public class TagUserAccountBroadCast {
    public long uid;
    public String mobile;
    public String user_key;
    public int updated_at;

    public static Logger LOG = LoggerFactory.getLogger(TagUserAccountBroadCast.class);
    public static final int MOBILELENGTH = 7;
    public static final String INPUT_KEY_BY = "uid";
    public static final String ENCRYPTMOBILE = "mobile";
    public static final String MOBILE_AREA_BROADCAST_DESC = "mobile_area_broadcast_desc";
    public static final int MOBILE_AREA_BROADCAST_WAIT_SEC = 5 * 1000; // 20*1000
    public static final String SPLITWORD = "#";

    public TagUserAccountBroadCast() {}

    public TagUserAccountBroadCast(long uid, String mobile, String user_key, int updated_at) {
        this.uid = uid;
        this.mobile = mobile;
        this.user_key = user_key;
        this.updated_at = updated_at;
    }

    public static TagUserAccountBroadCast init(long uid) {
        return new TagUserAccountBroadCast(uid, "", "", 0);
    }

    public static SingleOutputStreamOperator<String> getUserAccountProcessedStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserAccount.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserAccount>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserAccount.class);

        KafkaSource<String> mobileAreaSource =
                KafkaSource.<String>builder()
                        .setBootstrapServers(Stage2GlobalSetting.MOBILE_AREA_KFK_BOOTADD)
                        .setTopics(Stage2GlobalSetting.MOBILE_AREA_KFK_TOPIC)
                        .setGroupId(Stage2GlobalSetting.MOBILE_AREA_KFK_GRP_ID)
                        .setStartingOffsets(OffsetsInitializer.earliest())
                        .setValueOnlyDeserializer(new SimpleStringSchema())
                        .build();

        SingleOutputStreamOperator<String> mobileAreaInput =
                env.fromSource(mobileAreaSource, WatermarkStrategy.noWatermarks(), "mobileArea")
                        .uid("mobileArea_input");

        MapStateDescriptor<String, String> mobileAreaBroadcastDesc =
                new MapStateDescriptor<>(MOBILE_AREA_BROADCAST_DESC, String.class, String.class);

        BroadcastStream<String> moibleAreaBroadcastStream =
                mobileAreaInput.broadcast(mobileAreaBroadcastDesc);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserAccount>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserAccount> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagUserAccount-filter")
                        .keyBy(new TagUserAccountKeySelectorFunc())
                        .flatMap(new TagUserAccountFlatMapFunc())
                        .setParallelism(12)
                        .keyBy(
                                new KeySelector<String, Long>() {
                                    @Override
                                    public Long getKey(String value) throws Exception {
                                        return JSONObject.parseObject(value).getLong(INPUT_KEY_BY);
                                    }
                                })
                        .connect(moibleAreaBroadcastStream)
                        .process(new UserJoinMobileAreaProcessFunc())
                        .setParallelism(12);

        return out;
    }

    private static class TagUserAccountKeySelectorFunc
            implements KeySelector<Tuple2<Boolean, TableUserAccount>, Long> {
        @Override
        public Long getKey(Tuple2<Boolean, TableUserAccount> value) throws Exception {
            return value.f1.getUid();
        }
    }

    private static class TagUserAccountFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserAccount>, String> {
        ValueState<TagUserAccountBroadCast> userAccountVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userAccountVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TagUserAccountBroadCast>(
                                            "userAccountVState", TagUserAccountBroadCast.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserAccount> value, Collector<String> out)
                throws Exception {
            try {
                long uid = value.f1.getUid();
                String user_key = value.f1.getUser_key();
                String mobile = value.f1.getRegister_mobile();
                /** cacser 加密 */
                if (CaesarUtil.isEncrypted(mobile)) {
                    mobile = CaesarUtil.decode(mobile);
                }
                if (mobile.length() >= MOBILELENGTH) {
                    mobile = mobile.substring(0, MOBILELENGTH);
                }
                int updated_at = value.f1.getUpdated_at();

                TagUserAccountBroadCast tmpState = userAccountVState.value();
                TagUserAccountBroadCast curState =
                        tmpState == null ? TagUserAccountBroadCast.init(uid) : tmpState;

                if (curState.getUpdated_at() < updated_at) {
                    if (!curState.getMobile().equals(mobile)
                            || !curState.getUser_key().equals(user_key)) {
                        curState.mobile = mobile;
                        curState.user_key = user_key;
                        curState.updated_at = updated_at;

                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("uid", uid);
                        jsonObject.put("user_key", user_key);
                        jsonObject.put("mobile", mobile);

                        out.collect(jsonObject.toString());
                        userAccountVState.update(curState);
                    }
                }
            } catch (Exception e) {
                LOG.error("TagUserAccount FlatMap Exception: " + e.toString());
                LOG.error("TagUserAccount FlatMap Exception input: " + value);
                e.printStackTrace();
            }
        }
    }

    private static class UserJoinMobileAreaProcessFunc
            extends KeyedBroadcastProcessFunction<String, String, String, String> {
        public static final MapStateDescriptor<String, String> mobAreaBrodcstdesc =
                new MapStateDescriptor<>(MOBILE_AREA_BROADCAST_DESC, String.class, String.class);
        MapState<String, String> mobAreaWaitListMState;
        long startTime = 0L;
        //        long bStateSize = 0L;
        @Override
        public void open(Configuration parameters) throws Exception {
            mobAreaWaitListMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<String, String>(
                                            "MobAreaWaitListMState", String.class, String.class));
            startTime = System.currentTimeMillis();
        }

        @Override
        public void processElement(String value, ReadOnlyContext ctx, Collector<String> out)
                throws Exception {
            try {
                long timediff = (ctx.timerService().currentProcessingTime() - startTime);
                ReadOnlyBroadcastState<String, String> readBState =
                        ctx.getBroadcastState(mobAreaBrodcstdesc);
                if (timediff >= MOBILE_AREA_BROADCAST_WAIT_SEC) {
                    // real process element
                    if (readBState != null) {
                        if (!mobAreaWaitListMState.isEmpty()) {
                            mobAreaWaitListMState.clear();
                        }
                        ProcessData(value, readBState, out);
                    }
                } else {
                    LOG.warn(
                            "debug UserJoinMobileAreaProcessFunc processElement addinto wait List value: "
                                    + value);
                    mobAreaWaitListMState.put(INPUT_KEY_BY, value);
                    long now = ctx.timerService().currentProcessingTime();
                    ctx.timerService()
                            .registerProcessingTimeTimer(now + MOBILE_AREA_BROADCAST_WAIT_SEC);
                }

            } catch (Exception e) {
                LOG.error(
                        "UserJoinMobileAreaProcessFunc processElement exception: " + e.toString());
                LOG.error("UserJoinMobileAreaProcessFunc processElement error input: " + value);
                e.printStackTrace();
            }
        }

        /**
         * {"id":17278,"paragraph":"1317276","province_simple":"广东","city_county_simple":"河源","isp":"中国联通"}
         */
        @Override
        public void processBroadcastElement(String value, Context ctx, Collector<String> out)
                throws Exception {
            try {
                BroadcastState<String, String> writeBState =
                        ctx.getBroadcastState(mobAreaBrodcstdesc);
                if (value != null) {
                    JSONObject inputJson = JSONObject.parseObject(value);
                    String mobile = inputJson.getString("paragraph");
                    String province = inputJson.getString("province_simple");
                    String city = inputJson.getString("city_county_simple");

                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    writeBState.put(mobile, province + SPLITWORD + city);
                    //                    bStateSize += 1;
                }
            } catch (Exception e) {
                LOG.error(
                        "UserJoinMobileAreaProcessFunc processBroadcastElement exception: "
                                + e.toString());
                LOG.error(
                        "UserJoinMobileAreaProcessFunc processBroadcastElement error input: "
                                + value);
                e.printStackTrace();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
                throws Exception {
            super.onTimer(timestamp, ctx, out);
            try {
                LOG.debug(
                        "debug UserJoinMobileAreaProcessFunc onTimer triggered isEmpty: "
                                + mobAreaWaitListMState.isEmpty());
                if (!mobAreaWaitListMState.isEmpty()) {
                    String waitMsg = mobAreaWaitListMState.get(INPUT_KEY_BY);
                    ReadOnlyBroadcastState<String, String> readBState =
                            ctx.getBroadcastState(mobAreaBrodcstdesc);
                    ProcessData(waitMsg, readBState, out);
                    mobAreaWaitListMState.clear();
                }
                ctx.timerService().deleteProcessingTimeTimer(timestamp);
            } catch (Exception e) {
                LOG.error("UserJoinMobileAreaProcessFunc onTimer exception: " + e.toString());
                e.printStackTrace();
            }
        }

        private void ProcessData(
                String value,
                ReadOnlyBroadcastState<String, String> readBState,
                Collector<String> out)
                throws Exception {
            JSONObject inputJson = JSONObject.parseObject(value);
            String mobile = inputJson.getString(ENCRYPTMOBILE);
            String res = readBState.get(mobile);
            if (res != null) {
                String[] splits = res.split(SPLITWORD);
                if (splits.length == 2) {
                    inputJson.put("mob_prov", splits[0].equals("null") ? null : splits[0]);
                    inputJson.put("mob_city", splits[1].equals("null") ? null : splits[1]);
                    //                    LOG.warn("Debug UserJoinMobileAreaProcessFunc ProcessData
                    // bstateSize: " + bStateSize + " res: " +
                    // inputJson.toString(SerializerFeature.WriteMapNullValue));
                    out.collect(inputJson.toString(SerializerFeature.WriteMapNullValue));
                } else {
                    LOG.error("UserJoinMobileAreaProcessFunc ProcessData error mobile: " + mobile);
                }
            } else {
                LOG.warn("UserJMobAreaPrcFun ProcData mob : " + mobile);
            }
        }
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public long getUid() {
        return uid;
    }

    public String getUser_key() {
        return user_key;
    }

    public String getMobile() {
        return mobile;
    }
}
